package plgnet

import (
	"dgo/MyProj/PrjHttpUtils"
	"dgo/MyProj/PrjHttpWs"
	"dgo/goutils/utils"
	"dgo/plg/plgbase"
	"fmt"
	"gitee.com/ymofen/gobase/golog"
	"gitee.com/ymofen/supervalue"
)

/**
  该插件是基于websocket实现的一个订阅中心
      当对订阅中心进行订阅通道时(所有通道除外) 会对ws进行订阅,
      当收到(topic)主题数据时,通过ex(订阅中心发布)
      当插件SendKVs, 会通过ws进行发送

  2023-01-13 10:41:14:
     改成单线程发送
*/

type ProjWsCltExPlug struct {
	plgbase.PlgBase
	ex             *utils.SubscribeCenter
	plugId         string
	conn_n         int
	request_attach int
	wsclt          *PrjHttpWs.WsHttpClt
	wssubtag       int
	wsclterrcnt    int
	lastWarn       string
	monitor_send   *utils.MonitorSpeedSizeRec
	monitor_recv   *utils.MonitorSpeedSizeRec
}

func (this *ProjWsCltExPlug) innerOnWsRecvTopic(sender *PrjHttpWs.WsHttpClt, topic string, data interface{}) {
	this.ex.DispatchData(topic, sender, data, this)
	return
}

func (this *ProjWsCltExPlug) innerSendKVs(topic string, kv ...interface{}) error {
	if this.wsclt != nil {
		l := len(kv)
		if l == 1 {
			return this.wsclt.PublishMessage(topic, kv[0])
		}

		json := wrapper.NewSVObject()
		json.SetKeyValues(kv...)
		return this.wsclt.PublishMessage(topic, json)
	} else {
		return fmt.Errorf("WS连接尚未进行创建")
	}

}

func (this *ProjWsCltExPlug) SendKVs(topic string, kv ...interface{}) error {
	utils.DefaultWorkers().PostTaskFunc(this, func() {
		err := this.innerSendKVs(topic, kv...)
		if err != nil {
			this.lastWarn = fmt.Sprintf("[%s]推送异常:%s", utils.NowString(), err.Error())
			this.wsclterrcnt++
		}
	})
	return nil
}

func (this *ProjWsCltExPlug) DispatchData(channelid string, sender interface{}, data interface{}, tag interface{}) (err error) {
	err = this.SendKVs(channelid, data)
	return err
}

func (this *ProjWsCltExPlug) SubscribeChannel(channelid string, id interface{}, cb utils.SubscribeCallback) {
	this.ex.SubscribeChannel(channelid, id, cb)
}

func (this *ProjWsCltExPlug) UnsubscribeChannel(channelid string, id interface{}) bool {
	return this.ex.UnsubscribeChannel(channelid, id)
}

func (this *ProjWsCltExPlug) OnHttpStatus(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	json := wrapper.NewSVObject()
	json.SetKeyValues("send-info", fmt.Sprintf("%s, %s", this.monitor_send.SpeedSize(), this.monitor_send.SpeedIO()),
		"recv-info", fmt.Sprintf("%s, %s", this.monitor_recv.SpeedSize(), this.monitor_recv.SpeedIO()),
		"conn-n", this.conn_n,
		"pub-err-n", this.wsclterrcnt,
		"pub-worker", utils.DefaultWorkers().GroupSimpleStatus(this),
		"lastwarn", this.lastWarn,
		"ex", this.ex.StatusString(),
		"ws-host", fmt.Sprintf("%s,%s", this.wsclt.GetWsUrl(), this.wsclt.GetWsOrgUrl()),
		"ws-sub-topic", this.wsclt.GetSubscribleTopics(),
		"connected", this.wsclt != nil && this.wsclt.IsConnected(), "lastdis", this.wsclt.LastDisreason())
	PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.AsJSONString(false))
	return
}

func (this *ProjWsCltExPlug) OnHttpDo(rec *PlgHttpRec) (rval interface{}, errcode int, err error) {
	action := rec.QryAsInt("action", 0)
	if action == 1 {
		this.wsclt.BreakConnect(fmt.Sprintf("[%s]HttpDo请求断线!", utils.NowString()))
		rec.ResponseJSONValues("code", 0, "msg", "请求断线成功!")
	} else {
		json := wrapper.NewSVObject()
		json.SetKeyValues("action-1", "请求断线")
		PrjHttpUtils.JSON2TableRec(rec.RespW, "", json.String())
	}
	return
}

func (this *ProjWsCltExPlug) BreakConnect() {
	this.wsclt.BreakConnect(fmt.Sprintf("[%s]HttpDo请求断线!", utils.NowString()))
}

/*
**

	{
	  "id":"core_kpl_wsclt",
	  "type":"http.wsclt.ex",
	  "conf":{
		"url":"http://xxx.xx.xx:xxx/$appuuid$",
	    "request-attach":1,
	    "wssubtag":1   // 调用ws进行订阅, 默认为1
	   }
	}
*/
func (this *ProjWsCltExPlug) ReloadConf(conf *wrapper.SuperValue) error {

	if this.monitor_send == nil {
		this.monitor_send = utils.NewMonitorSpeedSizeRec()
		this.monitor_recv = utils.NewMonitorSpeedSizeRec()
		utils.Monitor.AddMonitorRec(this.monitor_send, this.monitor_send)
		utils.Monitor.AddMonitorRec(this.monitor_recv, this.monitor_recv)
	}

	conf.CheckExistsIntFunc("request-attach", func(val int64) {
		this.request_attach = int(val)
		if this.wsclt != nil {
			this.wsclt.RequestAttachInfoFlag = int(val)
		}
	})

	conf.CheckExistsIntFunc("wssubtag", func(val int64) {
		this.wssubtag = int(val)
	})

	conf.CheckExistsStrFunc("url", func(strv string) {
		strv = utils.ReplacePlaceholder(strv, "$$", "$$")
		strv = plgbase.ReplacePlaceholderForFileName(strv)
		if this.wsclt == nil {
			if len(strv) == 0 {
				return
			}
			wsclt := PrjHttpWs.NewWsHttpClt(strv, "")
			wsclt.RequestAttachInfoFlag = this.request_attach
			wsclt.ConfigOnRecvTopic(this.innerOnWsRecvTopic)
			wsclt.ConfigWsOnConnectErr(func(sender *PrjHttpWs.WsClient, err error) {
				golog.Warnf("[%d]连接服务器(%s)异常:%s", this.PlugDNA, sender.GetWsReqUrl(), err.Error())
			})
			wsclt.OnConnectedFunc = func(sender *PrjHttpWs.WsHttpClt) {
				this.conn_n++
			}
			wsclt.ConfigMonitor(this.monitor_send, this.monitor_recv)
			utils.GoFunc2(wsclt.Start)
			this.wsclt = wsclt
		} else {
			this.wsclt.ConfigWsReqUrl(strv, "")
		}
	})

	if this.ex == nil {
		this.ex = utils.NewSubscribeCenterASync()
		this.ex.OnSubChannel = func(channelid string, n int) {
			if channelid == "*" {
				return
			}
			if this.wssubtag == 1 {
				wsclt := this.wsclt
				if wsclt != nil {
					wsclt.SubScribleTopic(channelid) // 订阅过的不会再进行订阅
				}
			}
		}
		this.ex.OnUnSubChannel = func(channelid string, n int) {
			if n == 0 {
				if channelid == "*" {
					return
				}
				wsclt := this.wsclt
				if wsclt != nil {
					wsclt.UnsubScribleTopic(channelid)
				}
			}
		}
	}

	return nil
}

func (this *ProjWsCltExPlug) OnObjectNotify(sender interface{}, notifytype int, data interface{}) (rval interface{}, err error) {
	if notifytype == utils.NOTIFY_TYPE_SETCONF {
		if conf, ok := data.(*wrapper.SuperValue); ok {
			err = this.ReloadConf(conf)
		} else {
			err = fmt.Errorf("非法的JSON配置")
		}
	} else if notifytype == utils.NOTIFY_TYPE_RELEASE {
		this.ReleaseFlag = 1

		if this.wsclt != nil {
			this.wsclt.Stop()
		}
	}
	return
}

func CreateProjWsCltExPlugFunc(args ...interface{}) (rval interface{}, err error) {
	obj := &ProjWsCltExPlug{}
	obj.wssubtag = 1
	if len(args) > 0 {
		if str, ok := args[0].(string); ok {
			obj.plugId = str
		}
	}
	rval = obj
	return
}
